#pragma once

#include <iostream>
#include <string>
#include <vector>
#include <cassert>
#include <cstring>
#include <algorithm>
#include "log.hpp"
#include <netinet/in.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>       
#include <unistd.h>
#include <fcntl.h>
#include <functional>
#include <sys/epoll.h>
#include <unordered_map>
#include <stdlib.h>
#include <pthread.h>
#include <sys/eventfd.h>
#include <mutex>
#include <memory>
#include <thread>
#include <condition_variable>
#include <sys/timerfd.h>
#include <signal.h>

#define LISTEN_BACKLOG 10

class Buffer
{
public:
    Buffer(uint64_t capacity = 100)
        : _reader_idx(0)
        , _writer_idx(0)
        , _buffer(capacity)
    {}


    char* Begin()
    {
        return &(*_buffer.begin());
    }

    // 获取读pos
    char* ReadPos()
    {
        return (Begin() + _reader_idx);
    }

    // 获取写pos
    char* WriterPos()
    {
        return (Begin() + _writer_idx);
    }

    // 获取前沿空间
    uint64_t HeadSpaceSize()
    {
        return _reader_idx;
    }

    // 获取后沿空间
    uint64_t TailSpaceSize()
    {
        // std::cout << "_buffer.size() " << _buffer.size() << "_writer_idx "  << _writer_idx << std::endl;
        return _buffer.size() - _writer_idx;
    }

    // 获取可读空间    
    uint64_t ReadableSize()
    {
        return _writer_idx - _reader_idx; 
    }

    // 移动读指针
    void MoveReadOffset(uint64_t len)
    {
        assert(len <= ReadableSize());

        _reader_idx += len;
    }

    // 移动写指针
    void MoveWriteOffset(uint64_t len)
    {
        assert(len <= TailSpaceSize());

        _writer_idx += len;
    }

    // 确保写入空间
    void EnsureWriteSpace(uint64_t len)
    {
        if(len <= TailSpaceSize())
            return;
        else if(len <= TailSpaceSize() + HeadSpaceSize())
        {
            uint64_t cnt = ReadableSize();
            
            // 移动数据
            char* pos = ReadPos();
            std::copy(pos, pos + ReadableSize(), Begin());

            // 数据移动 更新指针
            _reader_idx = 0;
            _writer_idx = cnt;
        }
        else
        {
            _buffer.resize(_buffer.size() + len);
            // std::cout << "+++++++++扩容+++++++++++" << std::endl;
        }
    }

    // 写入数据
    void Write(const char* buf, uint64_t len)
    {
        EnsureWriteSpace(len);
        const char* pos = buf;
        std::copy(pos, pos + len, WriterPos());
    }

    void WriteAndPush(const char* buf, uint64_t len)
    {
        Write(buf, len);
        MoveWriteOffset(len);
    }

    void WriteAsString(const std::string& str)
    {
        Write(str.c_str(), str.size());
    }

    void WriteAsStringAndPush(const std::string& str)
    {
        // std::cout << str.size() << std::endl;

        WriteAsString(str);
        MoveWriteOffset(str.size());
    }

    void WriteBuffer(Buffer& data) 
    {
        return Write(data.ReadPos(), data.ReadableSize());
    }

    void WriteBufferAndPush(Buffer& data) 
    { 
        WriteBuffer(data);
        MoveWriteOffset(data.ReadableSize());
    }

    // 读数据
    void Read(void* buf, uint64_t len)
    {
        char* pos = (char*)ReadPos();
        std::copy(pos, pos + std::min(len, ReadableSize()), (char*)buf);
    }

    void ReadAndPop(char* buf, uint64_t len)
    {
        Read(buf, len);
        MoveReadOffset(len);
    }

    std::string ReadAsString(uint64_t len)
    {
        assert(len <= ReadableSize());

        std::string str;
        str.resize(len);
        Read(&(str[0]), len);

        return std::move(str);
    }

    std::string ReadAsStringAndPop(uint64_t len)
    {
        std::string ret = ReadAsString(len);
        MoveReadOffset(len);

        return ret;
    }

    // 找回车换行符
    char* FindCRLF()
    {
        // memchr 用于在指定内存块中查找某个字符（字节）的首次出现位置
        char* ret = (char*)memchr(ReadPos(), '\n', ReadableSize());
        return ret;
    }

    std::string GetLine()
    {
        char* ret = FindCRLF();
        if(ret == nullptr)
            return "";

        char* pos = ReadPos();
        // 加一为了取出换行符
        return ReadAsString(ret - pos + 1);
    }

    std::string GetLineAndPop()
    {
        char* ret = FindCRLF();
        if(ret == nullptr)
            return "";

        char* pos = ReadPos();
        // 加一为了取出换行符
        return ReadAsStringAndPop(ret - pos + 1);
    }

    void clear()
    {
        _reader_idx = 0;
        _writer_idx = 0;
    }

private:
    std::vector<char> _buffer;
    uint64_t _reader_idx;
    uint64_t _writer_idx;
};





class Socket
{
public:
    Socket()
        : _sockfd(-1)
    {}

    Socket(int fd)
        : _sockfd(fd)
    {}

    ~Socket()
    {
        Close();
    }

    int GetFd()
    {
        return _sockfd;
    }

    bool SockCreate()
    {
        _sockfd = socket(AF_INET, SOCK_STREAM, 0); //应该式这样的哈
        if(_sockfd < 0)
        {
            LOG(Error, "socket error, errno : %d, errno string : %s", errno, std::strerror(errno));
            exit(1);
        }
        ReuseAddress();
        return true;
    }

    bool Bind(const std::string& ip, uint16_t port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        // std::cout << port << std::endl;
        addr.sin_port = htons(port);
        // std::cout << addr.sin_port << std::endl;
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(addr);

        int ret = bind(_sockfd, (struct sockaddr*)&addr, len);
        if(ret < 0)
        {
            LOG(Error, "bind error, errno : %d, errno string : %s", errno, std::strerror(errno));
            exit(1);
        }
        LOG(Info, "bind success");
        return true;
    }

    bool Listen(int backlog = LISTEN_BACKLOG)
    {
        int ret = listen(_sockfd, backlog);
        if(ret < 0)
        {
            LOG(Error, "listen error, errno : %d, errno string : %s", errno, std::strerror(errno));
            return false;
        }

        LOG(Info, "listen success");
        return true;
    }

    int Accept()
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        socklen_t len = sizeof(addr);

        int fd = accept(_sockfd, (struct sockaddr*)&addr, &len);
        if(fd < 0)
        {
            LOG(Error, "Accept error, errno : %d, errno string : %s", errno, std::strerror(errno));
            return -1;
        }

        LOG(Info, "Accept success , fd : %d", fd);
        return fd;
    }

    bool Connect(uint16_t port, const std::string& ip)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(addr);
        
        int ret = connect(_sockfd, (struct sockaddr*)&addr, len);
        if(ret < 0)
        {
            LOG(Error, "connect error, errno : %d, errno string : %s", errno, std::strerror(errno));
            return false;
        }
        LOG(Info, "connect success");
        return true;
    }

    void Close()
    {
        if(_sockfd != -1)
        {
            close(_sockfd);
            _sockfd = -1;
        }
    }

    ssize_t Send(const char* buf, size_t len, int flags = 0)
    {
        ssize_t ret = send(_sockfd, buf, len, flags);

        if(ret <= 0) //这里少了个判断，少了个链接断开d判断，==0表示链接断开
        {
            if(errno == EAGAIN || errno == EINTR)
                return 0;
            
            LOG(Error, "send error, errno : %d, errno string : %s", errno, std::strerror(errno));
            return -1;
        }

        return ret;
    }

    ssize_t NonBlockSend(const char* buf, size_t len)
    {
        if(len == 0)
            return 0;
        return Send(buf, len, MSG_DONTWAIT);
    }

    ssize_t NonBlockRecv(char* buf, size_t len)
    {
        if(len == 0)
            return 0;

        // MSG_DONTWAIT 表示当前非阻塞接收
        return Recv(buf, len, MSG_DONTWAIT);
    }

    ssize_t Recv(char* buf, size_t len, int flags = 0)
    {
        int ret = recv(_sockfd, buf, len, flags);

        if(ret == 0)  // 读完了
        {
            return -1; //这里表示链接断开了，要返回-1，表示无法进行后续了该释放了，不能返回0
        }
        else if(ret < 0) // 出错了
        {
            // errno == EAGAIN 表示socket缓冲区中没有数据了，在非阻塞情况下才有
            // errno == EINTR  表示系统调用被中断
            if(errno == EAGAIN || errno == EINTR)
                return 0;
            
            LOG(Error, "recv error, errno : %d, errno string : %s", errno, std::strerror(errno));
            return -1;
        }

        return ret;
    }
    

    // 设置套接字选项--开启地址端口重用
    void ReuseAddress()
    {
        int val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void*)&val, sizeof(int));
        val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void*)&val, sizeof(int));
    }

    //设置套接字属性 非阻塞
    void NonBlock()
    {
        int fl = fcntl(_sockfd, F_GETFL, 0);
        fcntl(_sockfd, F_SETFD, fl | O_NONBLOCK);
    } 

    // 创建一个服务端
    bool CreateServer(uint16_t port, const std::string& ip = "0.0.0.0", bool flag = false)
    {
        if(SockCreate() == false)
            return false;
        if(Bind(ip, port) == false)
            return false;
        if(Listen() == false)
            return false;
        
        if(flag)
            NonBlock();

        

        return true;
    }

    // 创建一个客户端
    bool CreateClient(uint16_t port, const std::string& ip)
    {
        if(SockCreate() == false)
            return false;
        if(Connect(port, ip) == false)
            return false;

        return true;
    }

private:
    int _sockfd;
};

class Poller;
class EventLoop;

// 对于一个文件描述符进行监控
class Channel
{
public:
    using EventCallback = std::function<void()>;

    Channel(EventLoop* loop, int fd = -1)
        : _fd(fd)
        , _events(0)
        , _revents(0)
        , _loop(loop)
    {}

    int GetFd()
    {
        return _fd;
    }



    uint32_t GetEvents()
    {
        return _events;
    } 

    // 文件描述符是否 监控读
    bool ReadAble()
    {
        return (_events & EPOLLIN);
    }

    // 文件描述符是否 监控写
    bool WriteAble()
    {
        return (_events & EPOLLOUT);
    }

    // 更新节点
    void Update();

    // 移除节点
    void Remove();

    // 启动读  设置监控读
    void EnableRead()
    {
        _events |= EPOLLIN;
        Update();
    }

    // 启动写  设置监控写
    void EnableWrite()
    {
        _events |= EPOLLOUT;
        Update();
    }

    // 取消 监控读   
    void DisableRead()
    {
        _events &= ~EPOLLIN;
        Update();
    }

    // 取消 监控写
    void DisableWrite()
    {
        _events &= ~EPOLLOUT;
        Update();
    }

    // 全部取消
    void DisableAll()
    {
        _events = 0;
        Update();
    }

    //更新revent
    void SetRevents(uint32_t revents)
    {
        _revents = revents;
    }


    // 设置5个回调函数
    void SetReadCallback(EventCallback read_cb)
    {
        _read_cb = read_cb;
    }

    void SetWriteCallback(EventCallback write_cb)
    {
        _write_cb = write_cb;
    }  

    void SetErrorCallback(EventCallback error_cb)
    {
        _error_cb = error_cb;
    }

    void SetCloseCallback(EventCallback close_cb)
    {
        _close_cb = close_cb;
    }
    
    void SetEventCallback(EventCallback event_cb)
    {
        _event_cb = event_cb;
    }


    void HandleEvent()
    {
        
        // EPOLLRDHUP 对方关闭链接， 关闭之前要先把缓冲区的数据处理一下
        if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI))
        {
            if(_read_cb)
                _read_cb();
        }

        // 有可能释放链接的操作事件 一次只处理一个
        if((_revents & EPOLLOUT))
        {
            if(_write_cb)
                _write_cb();
        }
        else if(_revents & EPOLLERR)
        {    
            // 出错会断开链接 所以任何事件的回调要放在前面
            if(_error_cb)
                _error_cb();
        }
        else if(_revents & EPOLLHUP)
        {   
            if(_close_cb)
                _close_cb();
        }

        // 任意事件的回调  
        if(_event_cb)
            _event_cb();   // 刷新活跃度
    }

private:
    int _fd;
    uint32_t _events;   // 当前需要监控的事件
    uint32_t _revents;  // 当前链接触发的事件
    EventLoop* _loop;
    EventCallback _write_cb;
    EventCallback _read_cb;
    EventCallback _error_cb;
    EventCallback _close_cb;
    EventCallback _event_cb;  // 任意事件的回调函数
};


#define MAX_EPOLL_NUM 1024

class Poller
{
public:
    Poller()
    {
        _epfd = epoll_create(8888);
        if(_epfd < 0)
        {
            LOG(Fatal, "epoll_create error, errno : %d, errno string : %s", errno, strerror(errno));
            exit(1);
        }
        LOG(Info, "epoll_create success");
    }

private:

    void Update(Channel* channel, int op)
    {
        struct epoll_event event;
        int fd = channel->GetFd();
        event.data.fd = fd;
        event.events = channel->GetEvents();

        epoll_ctl(_epfd, op, fd, &event);
    }

public:

    void UpdateEvents(Channel* channel)
    {
        int fd = channel->GetFd();
        if(_channels.count(fd)) // 所以下次相同的fd的channel过来就不添加可读时间监控了
        {
            Update(channel, EPOLL_CTL_MOD);
            return;
        }
        
        _channels[fd] = channel;
        Update(channel, EPOLL_CTL_ADD);        
    }

    void RemoveEvents(Channel* channel)
    {
        int fd = channel->GetFd();
        if(_channels.count(fd) == 0)
        {
            LOG(Debug, "要删除的节点不存在,fd : %d", fd);
            return;
        }
        Update(channel, EPOLL_CTL_DEL);
        _channels.erase(fd); //没有从删除fd的数组 --- 就是这个问题了

        LOG(Info, "移除对fd : %d 的关心", fd);
    }

    // 监控 并 返回活跃链接
    void poll(std::vector<Channel*>& active)
    {
        int nfds = epoll_wait(_epfd, _evts, MAX_EPOLL_NUM, -1);
        if(nfds < 0)
        {
            if(errno == EINTR)
                return;
            
            LOG(Error, "epoll_wait error, errno : %d, errno string : %s", errno, strerror(errno));
            exit(1);
        }

        for(int i = 0; i < nfds; i++)
        {
            int fd = _evts[i].data.fd;
            uint32_t revents = _evts[i].events;

            _channels[fd]->SetRevents(revents);
            active.push_back(_channels[fd]);
        }
    }

private:

    int _epfd;
    struct epoll_event _evts[MAX_EPOLL_NUM];
    std::unordered_map<int, Channel*> _channels;
};



// 任务包装器
class TimerTask
{
public:
    using TaskFunc = std::function<void()>;
    using ReleaseFunc = std::function<void()>;
    
    TimerTask(uint64_t id, uint32_t timeout, const TaskFunc& cb)
        : _id(id)
        , _timeout(timeout)
        , _task_cb(cb)
        , _canceled(false)
    {}

    ~TimerTask()
    {
        if(_canceled == false)
            _task_cb();
        
        _release();
    }

    uint32_t GetDelay()
    {
        return _timeout;
    }

    void SetRelease(const ReleaseFunc& release)
    {
        _release = release;
    }

    void Cancel()
    {
        _canceled = true;
    }

private:
    uint64_t _id;        //任务id
    uint32_t _timeout;   //超时时间
    TaskFunc _task_cb;   //超时任务
    bool _canceled;      //任务是否取消 false--没有取消 true取消
    ReleaseFunc _release;     //把任务从时间轮删掉
};


// 时间轮
class TimeWheel
{
public:
    using PtrTask = std::shared_ptr<TimerTask>;
    using WeakTask = std::weak_ptr<TimerTask>;

    TimeWheel(EventLoop* loop, uint32_t tick = 0, int capacity = 100)
        : _tick(tick)
        , _capacity(capacity)
        , _wheel(_capacity)
        , _loop(loop)
        , _timerfd(CreateTimerfd())
        , _timer_channel(new Channel(loop, _timerfd))
    {
        _timer_channel->SetReadCallback(std::bind(&TimeWheel::Ontime, this));
        _timer_channel->EnableRead();
    }

private:
    static int CreateTimerfd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if(timerfd < 0)
        {
            LOG(Error, "timerfd_create error");
            exit(1);
        }

        struct itimerspec itime;

        // 第一次超时时间为1s后
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0;
        // 第一次超时后 每次超时时间
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 1;

        timerfd_settime(timerfd, 0, &itime, NULL);

        return timerfd;
    }

    void Remove(uint64_t id)
    {
        // 找
        if(_timer.count(id) == 0)
            return;
        
        // 删
        _timer.erase(id);
    }

    void TimerAddTaskLoop(uint64_t id, uint32_t timeout, const TimerTask::TaskFunc& cb)
    {

        PtrTask pt(new TimerTask(id, timeout, cb));
        pt->SetRelease(std::bind(&TimeWheel::Remove, this, id));
        
        WeakTask wt = pt;
        _timer[id] = wt;

        uint32_t pos = (_tick + timeout) % _capacity;
        _wheel[pos].push_back(pt);        
    }


    void TimerRefreshTaskLoop(uint64_t id)
    {
        if(_timer.count(id) == 0)
            return;
        
        PtrTask pt = _timer[id].lock();
        uint32_t pos = (_tick + pt->GetDelay()) % _capacity;
        _wheel[pos].push_back(pt);
    }

    void TimerCancelLoop(uint64_t id)
    {
        // 找
        if(_timer.count(id) == 0)
            return;

        //取消
        PtrTask pt = _timer[id].lock();
        if(pt)
            pt->Cancel();
    }


    // 读取 定时器描述符 的内容， 防止对下次产生影响
    int ReadTimerfd()
    {
        uint64_t times;
        int ret = read(_timerfd, &times, sizeof(times));
        if(ret < 0)
        {
            LOG(Error, "ReadTimerfd read error");
            exit(1);
        }
        return times;
    }

    void RunTimer()
    {
        _tick = (_tick + 1) % _capacity;

        _wheel[_tick].clear();
    }

    // timer_channel 的读回调函数-执行任务
    void Ontime()
    {
        int sz = ReadTimerfd();
        for(int i = 0 ; i < sz; i++)
        {
            RunTimer();
        }
    }

public:
    void TimerAddTask(uint64_t id, uint32_t timeout, const TimerTask::TaskFunc& cb);

    void TimerRefreshTask(uint64_t id);

    void TimerCancelTask(uint64_t id);

    bool Hastimer(uint64_t id)
    {
        if(_timer.count(id))
            return true;
        else
            return false;
    }

private:
    
    uint32_t _tick;      // 嘀嗒指针
    int _capacity;  // 容量
    std::vector<std::vector<PtrTask>> _wheel;
    std::unordered_map<uint64_t, WeakTask> _timer;

    int _timerfd;  // 定时器描述符--可读事件的回调函数就是 计数器执行定时任务
    EventLoop* _loop;
    std::unique_ptr<Channel> _timer_channel;
};

using Functor = std::function<void()>;

class EventLoop
{
public:
    
    EventLoop()
        : _thread_id(std::this_thread::get_id())
        , _event_fd(CreateEventFd())
        , _event_channel(new Channel(this, _event_fd))
        , _timer_wheel(this)
    {
        _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventfd,this));
        _event_channel->EnableRead();
    }

    // 判断任务是否在当前线程
    bool IsInLoop()
    {
        return _thread_id == std::this_thread::get_id();
    }

    bool AssertInloop()
    {
        assert(_thread_id == std::this_thread::get_id());
    }

    static int CreateEventFd()
    {
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if(efd < 0)
        {
            LOG(Error, "eventfd error, errno : %d, errno string : %s", errno, std::strerror(errno));
            exit(1);
        }

        return efd;
    }

    void ReadEventfd()
    {
        uint64_t res = 0;
        int ret = read(_event_fd, &res, sizeof(res));
        if(ret < 0)
        {
            // EINTR --  被信号打断
            // EAGAIN -- 无数据可读
            if(errno == EINTR || errno == EAGAIN)
            {
                return;
            }

            LOG(Error, "ReadEventfd error, errno : %d, errno string : %s", errno, std::strerror(errno));
            exit(1);
        }

        return;
    }

    // 唤醒
    // 唤醒就是像 eventfd 写入一个数据，epoll 知道了eventfd 的读事件就绪就不会继续阻塞了
    void WeadUpEventfd()
    {
        uint64_t val = 1;
        int ret = write(_event_fd, &val, sizeof(val));
        if(ret < 0)
        {
            if(errno == EINTR)
            {
                return;
            }

            LOG(Error, "WeadUpEventfd error, errno : %d, errno string : %s", errno, std::strerror(errno));
            exit(1);
        }

    }
    
    // 压入任务队列
    void QueueInLoop(const Functor& cb)
    {
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.push_back(cb);
        }

        /* 唤醒有可能因为没有IO事件就绪而导致的epoll阻塞*/

        WeadUpEventfd();
    }

    // 要执行的任务是否在当前线程， true-执行 false-压入任务队列
    void RunInLoop(const Functor& cb)
    {
        if(IsInLoop())
        {
            cb();
            return;
        }

        QueueInLoop(cb);
        return;
    }

    // 更新 文件描述符 要监听的 任务
    void UpdateEvents(Channel* channel)
    {
        _poller.UpdateEvents(channel);
    }

    // 删除 对这个文件描述符 的监听
    void RemoveEvents(Channel* channel)
    {
        _poller.RemoveEvents(channel);
    }

    // 把所有任务进行执行
    void RunAllTask()
    {
        std::vector<Functor> functor;
        {
            std::unique_lock<std::mutex> _lock(_mutex);
            _tasks.swap(functor);
        }

        for(auto& f : functor)
        {
            f();
        }
    }

    void Start()
    {
        while(1)
        {
            // 1, 事件监控
            std::vector<Channel*> active;
            _poller.poll(active);

            // 2, 事件处理
            for(auto& ch : active)
            {
                ch->HandleEvent();
            }

            // 3, 执行任务
            RunAllTask();
        }
    }

    // 时间轮 相关接口
    void TimerAdd(uint64_t id, uint32_t timeout, const TimerTask::TaskFunc& cb)
    {
        return _timer_wheel.TimerAddTask(id, timeout, cb);
    }

    void TimerRefresh(uint64_t id)
    {
        return _timer_wheel.TimerRefreshTask(id);
    }

    void TimerCancel(uint64_t id)
    {
        return _timer_wheel.TimerCancelTask(id);
    }

    bool HasTimer(uint64_t id)
    {
        return _timer_wheel.Hastimer(id);
    }


private:
    Poller _poller;
    int _event_fd;                 // 提醒防止epoll阻塞的
    std::thread::id _thread_id;    // 线程id
    std::unique_ptr<Channel> _event_channel;
    std::vector<Functor> _tasks;   // 任务池
    std::mutex _mutex;             // 保证任务池线程安全

    TimeWheel _timer_wheel;        // 时间轮
};


class LoopThread
{
public:
    LoopThread()
        : _loop(nullptr)
        , _thread(std::thread(&LoopThread::ThreadEntry, this))
    {}

    EventLoop* GetLoop()
    {
        EventLoop* loop = nullptr;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [&](){ return _loop != nullptr; });
            loop = _loop;
        }

        return loop;
    }
private:
    // 线程执行的函数
    void ThreadEntry()
    {
        EventLoop loop;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }

        loop.Start();
    }
private:
    EventLoop* _loop;
    std::thread _thread;
    std::mutex _mutex;
    std::condition_variable _cond;
};


class LoopThreadPool
{
public:
    LoopThreadPool(EventLoop* baseloop)
        : _baseloop(baseloop)
        , _thread_count(0)
        , _next_thread(0)
    {}

    void SetThreadCount(int count)
    {
        _thread_count = count;
    }

    void Create()
    {
        if(_thread_count > 0)
        {
            _threads.resize(_thread_count);
            _loops.resize(_thread_count);

            for(int i = 0; i < _thread_count; i++)
            {
                _threads[i] = new LoopThread();
                _loops[i] = _threads[i]->GetLoop();
            }
        }
    }

    EventLoop* GetNextLoop()
    {
        if(_thread_count == 0)
            return _baseloop;
        
        _next_thread = (_next_thread + 1) % _thread_count;
        
        return _loops[_next_thread];
    }

private:
    EventLoop* _baseloop;
    int _thread_count;
    int _next_thread;
    std::vector<LoopThread*> _threads;
    std::vector<EventLoop*> _loops;
};


class Any
{
private:
    class Holder
    {
    public:
        virtual ~Holder()
        {}

        virtual const std::type_info& GetType() = 0;

        virtual Holder* clone() = 0;
    };


    template <typename T>
    class PlaceHolder : public Holder
    {
    public:

        PlaceHolder(const T& val)
            : _val(val)
        {}

        virtual const std::type_info& GetType()
        {
            return typeid(T);
        }

        virtual Holder* clone()
        {
            return new PlaceHolder<T>(_val);
        }

        T _val;
    };

public:

    Any() : _content(nullptr) {};

    Any(const Any& other)
        : _content(other._content ? other._content->clone() : nullptr)
    {}

     ~Any()
    {
        if (_content)
        {
            delete _content;
        }
    }

    template<typename T>
    Any(const T& val)
        : _content(new PlaceHolder<T>(val))
    {}

    template <typename T>
    T* GetVal()
    {
        if(typeid(T) != _content->GetType())
        if(nullptr == _content)
            return nullptr;
        
        return &(((PlaceHolder<T>*)_content)->_val);
    }

    void swap(Any& other)
    {
        std::swap(_content, other._content);
    }

    Any& operator=(const Any& other)
    {
        Any(other).swap(*this);
        return *this;
    }

    template<typename T>
    Any& operator=(const T& val)
    {
        _content = new PlaceHolder<T>(val);
    }

    Holder* _content = nullptr;
};


typedef enum
{
    DISCONNECTED,    // 链接关闭状态
    CONNECTING,      // 链接建立成功 待处理
    CONNECTED,       // 已连接 各种设置已经完成 可以通信
    DISCONNECTING,   // 待关闭
}ConnStatu;

class Connection;

using PtrConnection = std::shared_ptr<Connection>;

class Connection : public std::enable_shared_from_this<Connection>
{
private:
    using ConnectCallback = std::function<void(const PtrConnection&)>;
    using MessageCallback = std::function<void(const PtrConnection&, Buffer*)>;
    using ClosedCallback = std::function<void(const PtrConnection&)>;
    using AnyEventCallback = std::function<void(const PtrConnection&)>;

    // 把要发送的数据发送到发送缓冲区
    void SendInloop (Buffer& buf)
    {
        if(_statu == DISCONNECTED)
            return;
        
        _out_buffer.WriteBufferAndPush(buf);

        // 开启对读事件的关心
        if(_channel.WriteAble() == 0)
        {
            _channel.EnableWrite();
        }
    }

    // 关闭链接 不是真的关闭， 要坚持是否还有数据待处理
    void ShutdownInloop()
    {
        // 改变链接状态
        _statu = DISCONNECTING;

        // 接收缓冲区还有数据就 处理一下业务
        if(_in_buffer.ReadableSize() > 0)
        {
            if(_message_cb)
            {
                _message_cb(shared_from_this(), &_in_buffer);
            }
        }
        // 发送缓冲区还有数据就发送一下
        if(_out_buffer.ReadableSize() > 0)
        {
            if(_channel.WriteAble() == 0)
            {
                _channel.EnableWrite();
            }
        }
        // 发送缓冲区数据发送完了 就真的关闭链接
        if(_out_buffer.ReadableSize() == 0)
        {
            Release();
        }
    }

    // 开启 非活跃事件销毁
    void EnableInactiveReleaseInloop(int sec)
    {
        _enable_inactive_release = true;
        // 存在定时任务 就延时销毁
        if(_loop->HasTimer(_conn_id))
        {
            return _loop->TimerRefresh(_conn_id);
        }
        // 不存在就新增定时任务
        _loop->TimerAdd(_conn_id, sec, std::bind(&Connection::Release, this));
    }

    // 关闭 非活跃事件销毁
    void CancelInactiveReleaseInloop()
    {
        _enable_inactive_release = false;
        if(_loop->HasTimer(_conn_id))
        {
            _loop->TimerCancel(_conn_id);
        }
    }

    // 切换协议 -- 非线程安全
    void UpgradeInLoop(const Any& context, ConnectCallback connected_cb, MessageCallback message_cb,
                        ClosedCallback closed_cb, AnyEventCallback any_event_cb)
    {
        _context = context;
        _connected_cb = connected_cb;
        _message_cb = message_cb;
        _closed_cb = closed_cb;
        _any_event_cb = any_event_cb;
    }

    // 正真的关闭链接 释放链接
    void ReleaseInloop()
    {
        // 修改链接状态
        _statu = DISCONNECTED;
        // 移除对该文件描述符的监控
        _channel.Remove();
        // close 关闭该文件描述符
        LOG(Info, "关闭文件描述符 close fd : %d", _sockfd);
        close(_sockfd);
        // 如果还有定时任务，就要销毁定时任务
        if(_loop->HasTimer(_conn_id))
        {
            CancelInactiveReleaseInloop();
        }

        // 调用关闭链接的回调
        // 用户
        if(_closed_cb)     
        {
            _closed_cb(shared_from_this());
        }
        //服务器
        if(_server_closed_cb)
        {
            _server_closed_cb(shared_from_this());
        }

    }

    void EstablishInloop()
    {
        // 修改链接状态
        assert(_statu == CONNECTING);
        _statu = CONNECTED;

        // 启动读事件
        _channel.EnableRead();

        // 调用创建好链接的回调
        if(_connected_cb)
        {
            _connected_cb(shared_from_this());
        }
    }

    // 文件描述符 触发 读事件
    void HandleRead()
    {
        // 1, 接收socket的数据, 并把数据放入缓冲区
        char buf[65536] = { 0 };
        int ret = _socket.NonBlockRecv(buf, 65535);
        if(ret < 0)
        {
            LOG(Debug, "读取数据出错 即将关闭链接");
            // 不能直接关闭链接， 要先检查缓冲区还有没有数据待处理
            ShutdownInloop();
            return;
        }
        else if(ret == 0)
        {
            LOG(Info, "缓冲区无数据");
            // ret == 0, 小错误，还有的救
            return;
        }
        // 将数据写入缓冲区
        _in_buffer.WriteAndPush(buf, ret);


        // 2, 调用message_callback 业务处理
        if(_in_buffer.ReadableSize() > 0)
        {
            // 继承 enable_shared_from_this<Connection>
            // 创建对象时 内部会得到一个这个对象的weak_ptr
            // shared_from_this 得到当前对象的一个shared_ptr
            if(_message_cb)
            {
                _message_cb(shared_from_this(), &_in_buffer);
            }
        }
    }

    // 文件描述符 触发 写事件
    void HandleWrite()
    {
        // 将发送缓冲区的内容发送到 接收方
        int ret = _socket.NonBlockSend((const char*)_out_buffer.ReadPos(), _out_buffer.ReadableSize());
        if(ret < 0)
        {
            // 发送错误就直接关闭链接， 这里是真的关闭链接
            if(_in_buffer.ReadableSize() > 0)
            {
                _message_cb(shared_from_this(), &_in_buffer);
   
            }
            Release();  // 关闭链接 实际释放
            return;
        }
        // 把数据发送后就要移动读指针
        _out_buffer.MoveReadOffset(ret);

        // 没有数据要发送了就要关闭对写事件的关心
        if(_out_buffer.ReadableSize() == 0)
        {
            _channel.DisableWrite();
            
            // 要是 链接状态为待关闭状态 且没有数据待发送 就关闭释放链接
            if(_statu == DISCONNECTING)
            {
                Release();
                return;
            }
        }
    }

    // 文件描述符 触发 挂断事件
    void HandleClose()
    {
        if(_in_buffer.ReadableSize() > 0)
        {
            _message_cb(shared_from_this(), &_in_buffer);
        }

        // 关闭链接 实际释放
        Release();
    }

    // 文件描述符 触发 错误事件
    void HandleError()
    {
        HandleClose();
    }

    // 文件描述符 触发 任意事件
    void HandleEvent()
    {
        // 刷新时间活跃度--延时关闭
        if(_enable_inactive_release == true)
        {
            _loop->TimerRefresh(_conn_id);
        }
        // 调用组件使用者的任意事件回调
        if(_any_event_cb)
        {
            _any_event_cb(shared_from_this());
        }

    }


public:
    Connection(EventLoop* loop, int sockfd, int conn_id)
        : _loop(loop)
        , _sockfd(sockfd)
        , _conn_id(conn_id)
        , _enable_inactive_release(false)
        , _socket(sockfd)
        , _channel(loop, sockfd)
        , _statu(CONNECTING)
    {
        _channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
        _channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
        _channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
    }

    ~Connection()
    {
        LOG(Info, "析构启动, Connection资源释放");
    }

    // 将数据发送到缓冲区
    void Send(const char* data, size_t len)
    {
        // 外界传入的data，可能是个临时的空间
        // 我们现在只是把发送操作压入了任务池，有可能并没有被立即执行
        // 因此有可能执行的时候，data指向的空间有可能已经被释放了
        Buffer buf;
        buf.WriteAndPush(data, len);
        _loop->RunInLoop(std::bind(&Connection::SendInloop, this, std::move(buf)));
    }

    // 关闭接口 需要判断有没有数据待处理
    void Shutdown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutdownInloop, this));
    }

    void Release()
    {
        _loop->QueueInLoop(std::bind(&Connection::ReleaseInloop, this));
    }

    // 启动非活跃销毁 并定义多长时间无通讯是非活跃
    void EnableInactiveRelease(int sec)     
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInloop, this, sec));
    }

    // 关闭非活跃销毁
    void CancelInactiveRelease()
    {
        _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInloop, this));
    }

    // 协议切换--重置上下文以及阶段性回调函数
    // 这个接口必须在Eventloop线程中立即执行
    // 防备新的事件触发后， 处理的时候，切换任务还没有被执行-导致数据用原协议处理
    void Upgrade(const Any& context,
                const ConnectCallback con_cb, const MessageCallback msg_cb,
                const ClosedCallback close_cb, const AnyEventCallback any_cb)
    {
        _loop->AssertInloop();
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, con_cb, msg_cb, close_cb, any_cb));
    }

    
    void Established()
    {
        _loop->RunInLoop(std::bind(&Connection::EstablishInloop, this));
    }

    int GetFd()
    {
        return _sockfd;
    }

    int GetId()
    {
        return _conn_id;
    }

    bool Connected() // 是否处于connected 状态
    {
        return _statu == CONNECTED;
    }

    void SetContext(const Any &context) 
    {
        _context = context; 
    }

    Any* GetContext()
    {
        return &_context;
    }

    void SetConnectCallback(const ConnectCallback& con_cb)
    {
        _connected_cb = con_cb;
    }

    void SetMessageCallback(const MessageCallback& msg_cb)
    {
        _message_cb = msg_cb;
    }

    void SetClosedCallback(const ClosedCallback& close_cb)
    {
        _closed_cb = close_cb;
    }

    void SetAnyEventCallback(const AnyEventCallback& any_cb)
    {
        _any_event_cb = any_cb;
    }

    void SetServerClosedCallback(const ClosedCallback& server_closed_cb)
    {
        _server_closed_cb = server_closed_cb;
    }

private:
    EventLoop* _loop;
    uint64_t _conn_id;  // 链接id & 任务id
    int _sockfd;
    Socket _socket;
    Channel _channel;
    Buffer _in_buffer;
    Buffer _out_buffer;
    Any _context;
    ConnStatu _statu;
    bool _enable_inactive_release = false;

    // 移除服务器的内部管理的链接信息
    ClosedCallback _server_closed_cb;

    // 用户自定义的回调函数
    ConnectCallback _connected_cb;
    MessageCallback _message_cb;
    ClosedCallback _closed_cb;
    AnyEventCallback _any_event_cb;
};


class Acceptor
{
public:
    using AcceptorCallback = std::function<void(int)>;

    Acceptor(EventLoop* loop, uint16_t port)
        : _loop(loop)
        , _socket(CreateServer(port))
        , _channel(_loop, _socket.GetFd())
    {
        _channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
    }

    void SetAcceptCallback(const AcceptorCallback& accept_cb)
    {
        _accept_cb = accept_cb;
    }

    void Listen()
    {
        _channel.EnableRead();
    }

private:
    int CreateServer(uint16_t port)
    {
        int ret = _socket.CreateServer(port);
        assert(ret == true);

        return _socket.GetFd();
    }

    void HandleRead()
    {
        int newfd = _socket.Accept();
        if(newfd < 0)
        {
            return;
        }

        if(_accept_cb)
        {
            _accept_cb(newfd);
        }
    }

    

private:
    Socket _socket;    // 创建监听套接字
    EventLoop* _loop;  
    Channel _channel;  // 对监听套接字进行事件管理
    AcceptorCallback _accept_cb;  // 新链接获取成功后的回调
};


class TcpServer
{
private:
    using ConnectCallback = std::function<void(const PtrConnection&)>;
    using MessageCallback = std::function<void(const PtrConnection&, Buffer*)>;
    using ClosedCallback = std::function<void(const PtrConnection&)>;
    using AnyEventCallback = std::function<void(const PtrConnection&)>;

    // 创建新链接
    void NewConnection(int fd)
    {
        _next_id++;
        PtrConnection conn(new Connection(_pool.GetNextLoop(), fd, _next_id));
        conn->SetConnectCallback(_connected_cb);
        conn->SetMessageCallback(_message_cb);
        conn->SetClosedCallback(_closed_cb);
        conn->SetAnyEventCallback(_any_event_cb);
        conn->SetServerClosedCallback(std::bind(&TcpServer::RemoveConnection, this, std::placeholders::_1));
        if(_enable_inactive_release)
        {
            // 启动非活跃超时销毁
            conn->EnableInactiveRelease(_timeout);
        }
        
        // 就绪初始化
        conn->Established();

        _conns.insert(std::make_pair(_next_id, conn));
    }

    // 从管理connection的_coons中移除链接
    void RemoveConnectionInloop(const PtrConnection& conn)
    {
        uint64_t id = conn->GetId();
        if(_conns.count(id))
        {
            _conns.erase(id);
        }
    }

    void RunAfterInloop(const Functor& task, int delay)
    {
        _next_id++;
        _baseloop.TimerAdd(_next_id, delay, task);
    }

public:
    TcpServer(uint16_t port)
        : _port(port)
        , _next_id(0)
        , _timeout(0)
        , _enable_inactive_release(false)
        , _acceptor(&_baseloop, port)
        , _pool(&_baseloop)
    {
        _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
        _acceptor.Listen();  // 开启事件连接管管理器 对读事件的关系
    }
    
    void SetThreadCount(int count)
    {
        _pool.SetThreadCount(count);
    }

    void RemoveConnection(const PtrConnection& conn)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInloop, this, conn));
    }

    // 设置延时任务
    void RunAfter(const Functor& task, int delay)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInloop, this, task, delay));
    }

    // 开启非活跃链接超时销毁
    void EnableInactiveRelease(int timeout)
    {
        _timeout = timeout;
        _enable_inactive_release = true;
    }

    void Start()
    {
        _pool.Create();      // 创建线程
        _baseloop.Start();
    }

    /*四个设置回调函数*/
    void SetConnectCallback(const ConnectCallback con_cb)
    {
        _connected_cb = con_cb;
    }

    void SetMessageCallback(const MessageCallback msg_cb)
    {
        _message_cb = msg_cb;
    }

    void SetClosedCallback(const ClosedCallback close_cb)
    {
        _closed_cb = close_cb;
    }

    void SetAnyEventCallback(const AnyEventCallback any_cb)
    {
        _any_event_cb = any_cb;
    }

private:
    uint16_t _port;
    uint64_t _next_id;      // 自动增长的链接id
    int _timeout;           
    bool _enable_inactive_release;
    EventLoop _baseloop;    // 主线程
    Acceptor _acceptor;     // 监听socket的管理对象
    LoopThreadPool _pool;   // 从属线程池
    std::unordered_map<uint64_t, PtrConnection> _conns;

    // 用户自定义的回调函数
    ConnectCallback _connected_cb;
    MessageCallback _message_cb;
    ClosedCallback _closed_cb;
    AnyEventCallback _any_event_cb;
};


void Channel::Update()
{
    _loop->UpdateEvents(this);
}

// 移除节点
void Channel::Remove()
{
    _loop->RemoveEvents(this);
}


void TimeWheel::TimerAddTask(uint64_t id, uint32_t timeout, const TimerTask::TaskFunc& cb)
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimerAddTaskLoop, this, id, timeout, cb));
}

void TimeWheel::TimerRefreshTask(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimerRefreshTaskLoop, this, id));
}

void TimeWheel::TimerCancelTask(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimeWheel::TimerCancelLoop, this, id));
}


class NetWork
{
public:
    NetWork()
    {
        LOG(Info, "SIGPIPE INIT");
        signal(SIGPIPE, SIG_IGN);
    }
};

static NetWork network;